import java.sql.{Connection, DriverManager, ResultSet}
import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{SparkConf, SparkContext}

object MySQLRDD {

    def main(args: Array[String]): Unit = {

        //1.创建spark配置信息
        val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("JdbcRDD")

        //2.创建SparkContext
        val sc = new SparkContext(sparkConf)

        //3.定义连接mysql的参数
        val driver = "com.mysql.jdbc.Driver"
        val url = "jdbc:mysql://127.0.0.1:3306/db0"
        val userName = "root"
        val passWd = "root"

        //创建JdbcRDD
        val sql: String = "select * from user where id>=? and id <= ?"
        val JDBCRDD = new JdbcRDD(sc, () => {
            Class.forName(driver)
            DriverManager.getConnection(url, userName, passWd)
        },
            sql,
            1,
            10,
            1,
            (r: ResultSet) => {
                println(r.getString("username") + "," + r.getString("PASSWORD"))
            }
        )
        JDBCRDD.collect()


        val rdd: RDD[(String, String)] = sc.makeRDD(List(("user1","password1"),("user2","password2")))
        rdd.foreachPartition(datas=>{
            Class.forName(driver)
            val conn: Connection = DriverManager.getConnection(url, userName, passWd)
            datas.foreach {
                case (username:String, password:String) =>{
                    val ps = conn.prepareStatement("insert into user(username,PASSWORD) values (?,?)")
                    ps.setString(1, username)
                    ps.setString(2, password)
                    ps.executeUpdate()
                    ps.close()
                    println(username+password)
                }
            }
            conn.close()
        })
        sc.stop()
    }
}
